package com.xiezc;

import com.alibaba.fastjson.JSON;
import com.xiezc.model.Device;
import com.xiezc.model.PTypeEnum;
import com.xiezc.model.SendProgress;
import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import static com.xiezc.Xutil.*;

/**
 * @author wb-xzc291800
 * @date 2018/09/05 15:35
 */

@Slf4j
public class XSync {

    Map<PTypeEnum, MsgHandlerInterface> msgHandler = new ConcurrentHashMap<>();

    /**
     * 设备的ID是由SnowFlake生成, 生成之后就不会改变
     */
    Map<Long, Device> deviceMap = new ConcurrentHashMap<>();

    Device self;

    private static SendPacketService sendPacketService;

    private static RecvPacketService recvPacketService;

    private static SendService sendService;

    private static RecvService recvService;

    private static RetPacketService retPacketService;

    private static AssembleMsgAndFile assembleMsgAndFile;

    private XSync() {

    }

    public static XSync create(Device self, List<Device> devices,
                               List<MsgHandlerInterface> msgHandlers) throws IOException {

        DatagramChannel channel = DatagramChannel.open();
        channel.socket().bind(self.getAddress());
        XSync xSync = new XSync();

        sendService = SendService.single(channel);
        recvService = RecvService.single(channel);
        sendPacketService = SendPacketService.single(xSync.deviceMap);
        recvPacketService = RecvPacketService.single(xSync.deviceMap);
        retPacketService = RetPacketService.single(waitMillSecond, maxSendCount);
        assembleMsgAndFile = AssembleMsgAndFile.single(recvPacketService, xSync.msgHandler);

        assembleMsgAndFile.setDeviceMap(xSync.deviceMap);
        for (Device device : devices) {
            xSync.deviceMap.put(device.getId(), device);
        }
        for (MsgHandlerInterface msgh : msgHandlers) {
            xSync.msgHandler.put(msgh.getPacketType(), msgh);
        }

        xSync.self = self;
        self.setConneted(true);
        self.setToken(Xutil.random.nextInt(10000));
        Xutil.selfDeviceId = self.getId();

        xSync.deviceMap.put(Xutil.selfDeviceId, self);
        log.warn("所有服务初始化完成.... ");

        sendPacketService.setSendService(sendService);
        sendPacketService.setRetPacketService(retPacketService);

        assembleMsgAndFile.setSendService(sendService);

        recvPacketService.setRecvService(recvService);
        recvPacketService.setRetPacketService(retPacketService);
        recvPacketService.setSendService(sendService);

        retPacketService.setSendService(sendService);

        log.warn("所有服务的互相依赖加载完成  .... ");

        return xSync;
    }

    public void start() {
        //启动组装和 处理消息的线程
        assembleMsgAndFile.start();
        log.info("组装和处理消息的线程启动 ... ");
        //启动接收消息的线程
        recvPacketService.start();
        log.info("接收消息的线程启动 ... ");
        //重发消息的线程启动
        retPacketService.start();
        log.info("重发消息的线程启动 ... ");
        //发送消息的底层线程启动
        sendService.start();
        log.info("发送消息的底层线程启动 ... ");
        //启动接收消息的底层线程
        recvService.start();
        log.info("接收消息的底层线程启动 ... ");
        //发送消息的线程启动
        sendPacketService.start();
        log.info("发送消息的线程启动 ... ");
        log.warn("启动完成.....");
    }

    /**
     * 发送登录消息.  只是发送登录消息,
     *
     * @param recv
     * @return
     */
    public SendProgress sendLogin(@NonNull Device recv, byte[] cont) throws InterruptedException {

        for (int i = 0; i < 5; i++) {
            if (recv.getAddress() != null) {
                IndexObj indexObj = new IndexObj(cont.length, Arrays.asList(recv)).invoke();
                sendOnePacket(cont, indexObj, PTypeEnum.Login);
                return indexObj.getSendProgress();
            } else {
                IndexObj indexObj = new IndexObj(cont.length, Arrays.asList(recv)).invoke();
                sendOnePacket(null, indexObj, PTypeEnum.Find_ret);
                Thread.sleep(1000);
            }
        }
        SendProgress sendProgress = new SendProgress();
        sendProgress.setSuccess(false);
        return sendProgress;

    }

    /**
     * 发送文件
     *
     * @param file
     * @param deviceIds
     * @return
     * @throws IOException
     */
    public SendProgress sendFile(File file, List<Long> deviceIds) throws IOException {

        if (!file.exists() || !file.isFile()) {
            throw new RuntimeException("要发送的文件不存在....");
        }

        List<Device> devices = new ArrayList<>();
        for (Long deviceId : deviceIds) {
            Device device = deviceMap.get(deviceId);
            if (device == null || !device.getConneted()) {
                log.warn("deviceId:{} 对应的设备不在线", deviceId);
            } else {
                devices.add(device);
            }
        }
        if (devices.isEmpty()) {
            throw new RuntimeException("消息的接收者全部不在线");
        }
        try (FileChannel fc = FileChannel.open(file.toPath())) {

            ByteBuffer byteBuffer = null;
            String fileName = file.getName();
            byte[] fileNameBytes = fileName.getBytes(Xutil.charset);
            //文件大小加上名称大小
            long fileLength = fc.size() + fileNameBytes.length + 4;

            IndexObj indexObj = new IndexObj(fileLength, devices).invoke();
            long len = indexObj.getPacketCount();

            for (int k = 0; k < len; k++) {
                //最后一个包的长度, 不一定就是packLen
                if (k == len - 1) {
                    byteBuffer = ByteBuffer.allocate((int)indexObj.getLastPacketLen() + fileNameBytes.length);
                    byteBuffer.putInt(fileNameBytes.length);
                    byteBuffer.put(fileNameBytes);
                } else if (k >= 0) {
                    byteBuffer = ByteBuffer.allocate(packetLen);
                }
                fc.read(byteBuffer);
                indexObj.setCurrent(k);
                sendOnePacket(byteBuffer.array(), indexObj, PTypeEnum.File);
            }
            return indexObj.getSendProgress();
        }
    }

    /**
     * 发送消息
     *
     * @param cont
     * @param deviceIds
     * @return
     */
    public SendProgress sendMsg(byte[] cont, List<Long> deviceIds) {

        List<Device> devices = new ArrayList<>();
        for (Long deviceId : deviceIds) {
            Device device = deviceMap.get(deviceId);
            if (device == null || !device.getConneted()) {
                log.warn("deviceId:{} 对应的设备不在线", deviceId);
            }
        }
        if (devices.isEmpty()) {
            throw new RuntimeException("消息的接收者全部不在线");
        }

        IndexObj indexObj = new IndexObj(Long.valueOf(cont.length), devices).invoke();
        ByteBuffer byteBuffer = ByteBuffer.wrap(cont);
        int currentLen = packetLen;
        long len = indexObj.getPacketCount();
        for (int k = 0; k < len; len++) {
            try {
                //最后一个包的长度, 不一定就是packLen
                if (k == len - 1) {
                    currentLen = (int)indexObj.getLastPacketLen();
                }
                byte[] packetCont = new byte[currentLen];

                byteBuffer.get(packetCont);
                indexObj.setCurrent(k);
                boolean sendSuccess = sendOnePacket(packetCont, indexObj, PTypeEnum.Msg);
                if (sendSuccess) {
                    log.error("阻塞队列加入对象失败 ,此条消息发送失败:{} ", JSON.toJSONString(indexObj));
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }

        return indexObj.getSendProgress();
    }

    /**
     * 发送一个包
     *
     * @param bytes
     * @return
     */

    private boolean sendOnePacket(byte[] bytes, IndexObj indexObj, PTypeEnum pTypeEnum) {

        try {

            SendPacket packet = new SendPacket();
            packet.setContent(bytes);
            if (bytes == null) {
                packet.setPackMd5(new byte[16]);
            } else {
                packet.setPackMd5(Xutil.md5(bytes));
            }
            packet.setIndex(indexObj.getCurrent());
            packet.setLen(indexObj.getPacketCount());

            packet.setToken(self.getToken());

            packet.setDeviceId(self.getId());

            packet.setPtype(pTypeEnum);
            packet.setMsgId(indexObj.getMsgId());
            packet.setDevices(indexObj.devices);

            packet.setSendProgress(indexObj.getSendProgress());
            return sendPacketService.add(packet);

        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return false;
    }

}

/**
 * 拆包信息的对象
 */
class IndexObj {

    @Getter
    SendProgress sendProgress;

    List<Device> devices;

    /**
     * 消息id
     */
    public long msgId;

    /**
     * 当前消息的index
     */
    public long current;

    /**
     * 文件或消息的总大小
     */
    private long size;

    /**
     * 拆成的包的数量
     */
    @Setter
    @Getter
    private long packetCount;

    /**
     * 最后一个包的长度
     */
    private long lastPacketLen;

    public SendProgress getSendProgress() {

        return sendProgress;
    }

    public IndexObj(long size, List<Device> devices) {

        this.size = size;
        this.devices = devices;
    }

    public IndexObj invoke() {

        msgId = Xutil.snowflake.generate();
        packetCount = size / packetLen;
        lastPacketLen = size % packetLen;
        if (lastPacketLen > 0) {
            packetCount = packetCount + 1;
        }
        current = 0;
        //新建进度条
        sendProgress = new SendProgress();
        sendProgress.setMsgId(this.msgId);
        sendProgress.setLen(packetCount);
        sendProgress.setStartTime(System.currentTimeMillis());
        return this;
    }

    public long getMsgId() {

        return msgId;
    }

    public void setMsgId(long msgId) {

        this.msgId = msgId;
    }

    public long getCurrent() {

        return current;
    }

    public void setCurrent(long current) {

        this.current = current;
    }

    public long getSize() {

        return size;
    }

    public void setSize(long size) {

        this.size = size;
    }

    public long getPacketCount() {

        return packetCount;
    }

    public void setPacketCount(long packetCount) {

        this.packetCount = packetCount;
    }

    public long getLastPacketLen() {

        return lastPacketLen;
    }

    public void setLastPacketLen(long lastPacketLen) {

        this.lastPacketLen = lastPacketLen;
    }

    public List<Device> getDevices() {

        return devices;
    }

    public void setDevices(List<Device> devices) {

        this.devices = devices;
    }
}